package cas.ihep.spark.rdd;

import cas.ihep.spark.util.ManagedResource;
import cas.ihep.spark.util.ManagedResourceKV;
import cas.ihep.spark.util.function.Function2;
import scala.Tuple2;
import scala.reflect.ClassTag;

import java.io.Serializable;
import java.net.SocketAddress;
import java.util.Map;

final class LoopBroadcastParam<T,K, Receive extends Serializable> implements Serializable {
    Receive zero;
    SocketAddress serverAddr;
    Function2<ManagedResource<T>,K, Receive> client;
    Function2<Receive, Receive, Receive> aggregate;
    ClassTag<K> classTag;
    Map<String,String> props;

    LoopBroadcastParam(SocketAddress addr, Receive z, Function2<Receive, Receive, Receive> a,
                       Function2<ManagedResource<T>,K, Receive> c, ClassTag<K> tag){
        serverAddr=addr;
        zero=z;
        aggregate=a;
        client=c;
        classTag=tag;
    }
    static <KK,VV, Send extends Serializable, Receive extends Serializable> LoopBroadcastParam<Tuple2<KK,VV>, Send, Receive> from(
            SocketAddress addr, Receive z, Function2<Receive, Receive, Receive> a, Function2<ManagedResourceKV<KK,VV>, Send, Receive> c, ClassTag<Send> tag){
        //noinspection unchecked
        return new LoopBroadcastParam<Tuple2<KK,VV>, Send, Receive>(addr,z,a,(Function2)c,tag);
    }
}